package com.atguigu.gmall.realtime.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.util.List;

/**
 * Author: Felix
 * Date: 2021/12/1
 * Desc: 维度查询的工具类
 */
public class DimUtil {

    public static JSONObject getDimInfo(String tableName, String id) {
        return getDimInfo(tableName,Tuple2.of("ID",id));
    }
    /**
     * @param tableName
     * @param columnNameAndValues
     * @return
     * 查询维度数据  使用旁路缓存的方式进行优化
     *      -先从Redis缓存中查询维度数据
     *      -如果缓存中有维度数据，缓存命中，直接将维度数据转换为jsonObj进行返回
     *      -如果缓存中没有查到维度数据，发送请求到phoenix数据库中查询
     *      -并将查询的结果放到Redis中缓存起来
     *
     * Redis进行缓存：
     *      type:String     key:  dim:维度表名:主键值1_主键值2    TTL:1day
     *
     *   getDimInfo("DIM_USER_INFO", Tuple2.of("ID","10"),Tuple2.of("NAME","zs"))
     */
    public static JSONObject getDimInfo(String tableName, Tuple2<String, String>... columnNameAndValues) {
        //拼接查询SQL语句
        StringBuilder selectSql = new StringBuilder("select * from " + tableName + " where ");
        //拼接操作Redis的key
        StringBuilder redisKey = new StringBuilder("dim:"+tableName.toLowerCase()+":");
        //遍历所有的查询条件参数
        for (int i = 0; i < columnNameAndValues.length; i++) {
            //每遍历一次，获取一组查询条件
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            //获取查询的列名
            String columnName = columnNameAndValue.f0;
            //获取查询的列值
            String columnValue = columnNameAndValue.f1;

            //将列名和值拼接到查询条件上
            selectSql.append(columnName + "='" + columnValue + "' ");
            //将值拼接到操作Redis的key上
            redisKey.append(columnValue);

            if(i < columnNameAndValues.length - 1){
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }

        //声明操作Redis的客户端
        Jedis jedis = null;
        //声明一个字符串  用于接收从Redis中查询出来的维度数据
        String jsonStr = null;
        //声明一个json对象，用于封装最后返回的维度信息
        JSONObject dimJsonObj = null;


        try {
            jedis = RedisUtil.getJedis();
            System.out.println("从Redis中查询数据的key是:" + redisKey);
            jsonStr = jedis.get(redisKey.toString());
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("从Redis中查询维度数据发生了异常");
        }

        //判断是否从Redis中查询查询出了维度数据
        if(jsonStr != null && jsonStr.length() > 0){
            //缓存命中    直接将缓存的内容转换为json对象
            dimJsonObj = JSON.parseObject(jsonStr);
        }else{
            //没有从Redis中查到数据，发送请求到phoenix表中查询
            System.out.println("从phoenix中查询维度的SQL语句为:" + selectSql);

            //底层还是调用的phoenixUtil中的查询方法，从phoenix表中查询数据
            List<JSONObject> jsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
            if(jsonObjList != null && jsonObjList.size() >0){
                //注意：因为我们是根据维度的主键去查询，所以如果存在维度数据的话，那么也只会有一条
                dimJsonObj = jsonObjList.get(0);

                //将从Phoenix中查询出来的数据放到Redis中缓存起来
                if(jedis != null){
                    jedis.setex(redisKey.toString(),3600*24,dimJsonObj.toJSONString());
                }
            }else{
                System.out.println("维度数据没找到:" + selectSql);
            }
        }

        //释放资源
        if(jedis != null){
            System.out.println("---关闭Jedis---");
            jedis.close();
        }


        return dimJsonObj;
    }

    //查询维度信息  没有优化  select * from DIM_BASE_TRADEMARK where id='13' and tm_name='bb';
    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String, String>... columnNameAndValues) {
        //拼接查询SQL语句
        StringBuilder selectSql = new StringBuilder("select * from " + tableName + " where ");
        //遍历所有的查询条件参数
        for (int i = 0; i < columnNameAndValues.length; i++) {
            //每遍历一次，获取一组查询条件
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            //获取查询的列名
            String columnName = columnNameAndValue.f0;
            //获取查询的列值
            String columnValue = columnNameAndValue.f1;

            //将列名和值拼接到查询条件上
            selectSql.append(columnName + "='" + columnValue + "' ");

            if(i < columnNameAndValues.length - 1){
                selectSql.append(" and ");
            }
        }
        System.out.println("从phoenix中查询维度的SQL语句为:" + selectSql);

        //定义一个json对象，用于封装查询的维度信息
        JSONObject dimJsonObj = null;
        //底层还是调用的phoenixUtil中的查询方法，从phoenix表中查询数据
        List<JSONObject> jsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
        if(jsonObjList != null && jsonObjList.size() >0){
            //注意：因为我们是根据维度的主键去查询，所以如果存在维度数据的话，那么也只会有一条
            dimJsonObj = jsonObjList.get(0);
        }else{
            System.out.println("维度数据没找到:" + selectSql);
        }

        return dimJsonObj;
    }

    public static  void deleteCached(String tableName,String id){
        try {
            String redisKey = "dim:"+tableName.toLowerCase()+":" + id;
            Jedis jedis = RedisUtil.getJedis();
            jedis.del(redisKey);
            jedis.close();
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("从Redis中删除缓存数据发生了异常");
        }
    }
    public static void main(String[] args) {
        //JSONObject jsonObj = getDimInfoNoCache("DIM_BASE_TRADEMARK", Tuple2.of("id", "13"),Tuple2.of("tm_name","bb"));
        JSONObject jsonObj = getDimInfo("DIM_BASE_TRADEMARK", Tuple2.of("id", "13"));
        System.out.println(jsonObj);
    }
}